package my.threading;

import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ExecutorExamples
{
	BlockingQueue<Runnable> queue;

	public static void main(String [] args) throws InterruptedException
	{
		System.out.println("Processors: " + Runtime.getRuntime().availableProcessors());
		ExecutorExamples test = new ExecutorExamples();
		test.runCanned();
		
	}
	
	public void runExplicit() throws InterruptedException
	{
		// when you use the default executor service you can't get at the queue
		queue = new LinkedBlockingQueue<Runnable>();
		
		Executor ex = new ThreadPoolExecutor(100, 200, 1, TimeUnit.MINUTES, queue, new TF());
		Timer t = new Timer();
		t.scheduleAtFixedRate(new Task(), new Date(), 1000);		
		
		for (int i = 0; i < 400; i++)
		{	
			ex.execute(new TestTask());
		}
		
		while (queue.size() != 0)
		{
			Thread.sleep(1000);
		}
		System.exit(0);
	}
	
	public void runCanned() throws InterruptedException
	{
		// when you use the default executor service you can't get at the queue
		queue = new LinkedBlockingQueue<Runnable>();		
		ExecutorService ex = Executors.newFixedThreadPool(100, new TF());
		
		// NOTE unbounded pool grows and collapses as needed
		//ex = Executors.newCachedThreadPool(new TF());
		Timer t = new Timer();
		t.scheduleAtFixedRate(new Task(), new Date(), 1000);		
		
		for (int i = 0; i < 400; i++)
		{	
			queue.add(new  TestTask());
		}
		
		for (int i = 0; i < 400; i++)
		{	
			ex.execute(new TestTask2());
		}
		
		
		while (queue.size() != 0)
		{
			Thread.sleep(1000);
		}
		System.exit(0);
	}
	
	class TF implements ThreadFactory {
	    AtomicInteger count = new AtomicInteger(1);
	    public synchronized Thread newThread(Runnable r) {
	        Thread t = new Thread(r) ;
	        t.setName("ThreadPool Thread[" + count.getAndIncrement() + "]");
	        t.setUncaughtExceptionHandler(new UEHLogger());
	      
	        return t;
	    }
	}
	
	class UEHLogger implements Thread.UncaughtExceptionHandler
    {
		public void uncaughtException(Thread t, Throwable e) {
			System.out.println(t + " threw exception: " + e);
		}
    }

	
	class Task extends TimerTask
	{

		@Override
		public void run()
		{
			System.out.println("Queue size: " + queue.size());
			
		}
		
	}
	
	class TestTask implements Runnable
	{

		public void run()
		{
			System.out.println("I'm " + Thread.currentThread().getName() + " with id: " + Thread.currentThread().getId());
			try
			{
				if((Thread.currentThread().getId() % 7) != 0)
				{
					Thread.sleep(1000);
				}
				else
				{
					throw new IllegalArgumentException("No sevens please please");
				}
			} catch (InterruptedException e)
			{
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
	}
	
	class TestTask2 implements Runnable
	{
		public void run()
		{
			System.out.println("I'm " + Thread.currentThread().getName() + " with id: " + Thread.currentThread().getId());			
			queue.poll();	
			try
			{
				Thread.sleep(1000);
			} catch (InterruptedException e)
			{
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
	}
}
